import tushare as ts
import datetime as dt
import pandas as pd
from datetime import datetime
from django.conf import settings
import logging,time,asyncio
from asgiref.sync import sync_to_async
from .models import *
from .constants import *
from django.db.models import Max
from django.db import transaction

logger = logging.getLogger(__name__)

# 创建异步的数据库操作函数
update_data_record = sync_to_async(DataUpdateRecord.update_record)

# 定义一个同步函数来获取所有股票列表（排除ST股票）
def get_stocks_list():
    return list(StockBasic.objects.all())

# 将同步函数转换为异步函数
get_all_stocks = sync_to_async(get_stocks_list)
def get_last_update_date(data_type):
    """
    获取指定数据类型的最后更新日期
    :param data_type: 数据类型，来自DATA_TYPES常量
    :return: 最后更新日期，如果没有记录则返回START_KDATA_DATE
    """
    
    if data_type == 'index_weekly':
        start_date =  START_KDATA_DATE_WEEK
    elif data_type == 'index_monthly':
        start_date = START_KDATA_DATE_MONTH
    else:
        start_date = START_KDATA_DATE
    try:
        record = DataUpdateRecord.objects.filter(data_type=data_type).order_by('-last_update_date').first()
        return record.last_update_date if record else start_date
    except Exception as e:
        logger.error(f"Error getting last update date for {data_type}: {str(e)}")
        return start_date

# 计算两个日期之间的交易日数量
def count_trading_days(start_date, end_date):
    """
    计算两个日期之间的交易日数量
    :param start_date: 开始日期，datetime.date对象或START_KDATA_DATE常量
    :param end_date: 结束日期，datetime.date对象
    :return: 交易日数量
    """
    if not start_date or not end_date:
        return 0
    
    # 如果start_date是START_KDATA_DATE常量，直接使用它（因为它已经是datetime.date对象）
    if start_date == START_KDATA_DATE:
        start_date = START_KDATA_DATE
    
    if start_date > end_date:#起始日期和结束日期相同时可能为一天
        return 0
    
    try:
        # 将日期转换为字符串格式YYYYMMDD
        start_str = start_date.strftime('%Y%m%d')
        end_str = end_date.strftime('%Y%m%d')
        
        # 查询交易日历
        trading_days = TradeDate.objects.filter(
            cal_date__gte=start_str,
            cal_date__lte=end_str,
            is_open=True
        ).count()
        
        return trading_days
    except Exception as e:
        logger.error(f"Error counting trading days: {str(e)}")
        return 0

# 将同步函数转换为异步函数
get_last_update_date_async = sync_to_async(get_last_update_date)
count_trading_days_async = sync_to_async(count_trading_days)
#为了避免tushareapi接口过于庞大，将tushareapi接口拆分成多个文件
#本接口包括了tushare接口的特色数据、两融数据、资金流向数据、参考数据、公募基金数据
class TushareAPI_stat:
    def __init__(self, token=None):
        """
        初始化 Tushare API
        :param token: Tushare API token，如果未提供则从环境变量获取
        """
        self.token = token or getattr(settings, 'TUSHARE_TOKEN', None)
        if not self.token:
            raise ValueError("Tushare API token is required")
        
        # 初始化 Tushare API
        ts.set_token(self.token)
        self.pro = ts.pro_api()
        self.plate_to_create = []#当前更新的板块列表，作为全局变量用于更新板块的个股关系数据

    async def update_all_data(self, start_date=None, end_date=None, websocket=None):
        """
        更新所有数据
        :param start_date: 开始日期，可选
        :param end_date: 结束日期，可选
        :param websocket: WebSocket对象，可选
        """
        results = {}
        try:
            #但实际更新还是看数据库中存储的各类数据的更新日期
            logger.info(f"Starting update of all market data from {start_date} to {end_date}")
            # 转换日期为datetime.date对象，用于计算交易日
            start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date() if start_date else dt.date.today()
            end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date() if end_date else dt.date.today()
            
            #################################首先计算那些数据需要更新###############################################
            # 计算各数据类型的实际起始日期
            def get_actual_start_date(last_update_date):
                if not last_update_date:
                    return start_date_obj
                else:#以数据库中记录的各个表格时间为准
                    next_day = last_update_date + dt.timedelta(days=1)
                    return  next_day
            #1. 检查个股筹码分布数据是否需要更新（每日更新）
            stock_chip_dist_last_update = await get_last_update_date_async(DATA_TYPES['STOCK_CHIP_DIST'])
            stock_chip_dist_start_date = get_actual_start_date(stock_chip_dist_last_update)
            stock_chip_dist_trade_days = await count_trading_days_async(stock_chip_dist_start_date, end_date_obj)

            #2. 检查市场、个股融资融券数据是否需要更新（每日更新）
            stock_margin_last_update = await get_last_update_date_async(DATA_TYPES['MARKET_STOCK_RZRQ'])
            stock_margin_start_date = get_actual_start_date(stock_margin_last_update)
            stock_margin_trade_days = await count_trading_days_async(stock_margin_start_date, end_date_obj)   

            #3. 检查个股资金流向数据是否需要更新（每日更新）
            stock_money_flow_last_update = await get_last_update_date_async(DATA_TYPES['STOCK_MONEYFLOW'])
            stock_money_flow_start_date = get_actual_start_date(stock_money_flow_last_update)
            stock_money_flow_trade_days = await count_trading_days_async(stock_money_flow_start_date, end_date_obj) 

            #4. 检查个股资金流向数据是否需要更新（每日更新）
            stock_money_flow_ths_last_update = await get_last_update_date_async(DATA_TYPES['STOCK_MONEYFLOW_THS'])
            stock_money_flow_ths_start_date = get_actual_start_date(stock_money_flow_ths_last_update)
            stock_money_flow_ths_trade_days = await count_trading_days_async(stock_money_flow_ths_start_date, end_date_obj) 

            #5. 检查板块（概念、行业）资金流向数据是否需要更新（每日更新）
            block_money_flow_ths_last_update = await get_last_update_date_async(DATA_TYPES['BLOCK_MONEYFLOW_THS'])
            block_money_flow_ths_start_date = get_actual_start_date(block_money_flow_ths_last_update)
            block_money_flow_ths_trade_days = await count_trading_days_async(block_money_flow_ths_start_date, end_date_obj)     
            
            #6.检查市场大盘资金流向数据是否需要更新（每日更新）
            market_money_flow_dc_last_update = await get_last_update_date_async(DATA_TYPES['MARKET_MONEYFLOW_DC'])
            market_money_flow_dc_start_date = get_actual_start_date(market_money_flow_dc_last_update)
            market_money_flow_dc_trade_days = await count_trading_days_async(market_money_flow_dc_start_date, end_date_obj) 

            # 设置总步骤数（包括同花顺相关数据和指数数据的更新）
            total_steps = (1 if stock_chip_dist_trade_days > 0 else 0) \
                + (1 if stock_margin_trade_days > 0 else 0) \
                + (1 if stock_money_flow_trade_days > 0 else 0) \
                + (1 if stock_money_flow_ths_trade_days > 0 else 0) \
                + (1 if block_money_flow_ths_trade_days > 0 else 0) \
                + (1 if market_money_flow_dc_trade_days > 0 else 0)
            # 初始化当前步骤数
            current_step = 0
            if websocket:
                await websocket.send_progress(
                    current=0,
                    total=total_steps,
                    message="开始下载数据..."
                )
            ########################################开始正式下载数据##############################################
            
            #1.更新个股筹码分布数据
            if stock_chip_dist_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新个股筹码分布数据... (距上次更新: {stock_chip_dist_trade_days}个交易日)"
                    )

                # 使用批处理机制更新个股筹码分布数据
                results['stock_chip_dist'] = await self.batch_update_stock_chip_dist(
                    start_date=stock_chip_dist_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated stock chip distribution data for {stock_chip_dist_trade_days} trading days from {stock_chip_dist_start_date}")
            #2.更新市场和个股融资融券数据
            if stock_margin_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新市场和个股融资融券数据... (距上次更新: {stock_margin_trade_days}个交易日)"
                    )                
                # 使用批处理机制更新市场和个股融资融券数据
                results['stock_margin'] = await self.batch_update_stock_margin(
                    start_date=stock_margin_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated stock margin data for {stock_margin_trade_days} trading days from {stock_margin_start_date}")

            #3.更新个股资金流向数据
            if stock_money_flow_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新个股资金流向数据... (距上次更新: {stock_money_flow_trade_days}个交易日)"
                    )
                # 使用批处理机制更新个股资金流向数据
                results['stock_money_flow'] = await self.batch_update_stock_money_flow(
                    start_date=stock_money_flow_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated stock money flow data for {stock_money_flow_trade_days} trading days from {stock_money_flow_start_date}")

            #4.更新同花顺个股资金流向数据
            if stock_money_flow_ths_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新个股资金流向数据... (距上次更新: {stock_money_flow_ths_trade_days}个交易日)"
                    )
                # 使用批处理机制更新个股资金流向数据
                results['stock_money_flow'] = await self.batch_update_stock_money_flow_ths(
                    start_date=stock_money_flow_ths_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated stock money flow data for {stock_money_flow_ths_trade_days} trading days from {stock_money_flow_ths_start_date}")

            #5.更新板块资金流向数据
            if block_money_flow_ths_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新板块资金流向数据... (距上次更新: {block_money_flow_ths_trade_days}个交易日)"
                    )
                # 使用批处理机制更新板块资金流向数据
                results['block_money_flow'] = await self.batch_update_block_money_flow(
                    start_date=block_money_flow_ths_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated block money flow data for {block_money_flow_ths_trade_days} trading days from {block_money_flow_ths_start_date}")
            #6.更新市场资金流向数据
            if market_money_flow_dc_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新市场资金流向数据... (距上次更新: {market_money_flow_dc_trade_days}个交易日)"
                    )
                # 使用批处理机制更新市场资金流向数据
                results['market_money_flow'] = await self.batch_update_market_money_flow(
                    start_date=market_money_flow_dc_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated market money flow data for {market_money_flow_dc_trade_days} trading days from {market_money_flow_dc_start_date}")

           
            # 更新数据更新记录（无论是否有更新都记录）
            await update_data_record(
                DATA_TYPES['ALL_DATA_2'],
                end_date_obj
            )


            if websocket:
                await websocket.send_progress(
                    current=total_steps,
                    total=total_steps,
                    message="所有基本数据数据2更新完成"
                )

            logger.info(f"Completed update of all market data: {results}")
            return results
        except Exception as e:
            logger.error(f"Error updating all market data: {str(e)}")
            raise
    async def batch_update_stock_chip_dist(self, start_date=None, end_date=None, websocket=None):
        """
        批量更新个股筹码分布数据
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :param websocket: WebSocket对象，可选
        :return: 更新的记录数
        """
        logger.info(f'Starting update of stock chip distribution data from {start_date} to {end_date}')
        try:
            # 解析日期
            if start_date and end_date:
                if '-' in start_date:  # 格式为YYYY-MM-DD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                # 如果没有指定日期范围，默认更新最近10个交易日
                end_date_obj = dt.date.today()
                start_date_obj = end_date_obj - dt.timedelta(days=20)  # 多取几天以确保包含10个交易日

            # 获取交易日列表
            trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
                cal_date__gte=start_date_obj.strftime('%Y%m%d'),
                cal_date__lte=end_date_obj.strftime('%Y%m%d'),
                is_open=True
            ).order_by('cal_date').values_list('cal_date', flat=True))

            if not trade_dates:
                logger.warning("No trading days found in the specified date range")
                return 0

            # 初始化计数器
            count_all = 0

            # 按每10个交易日分批处理
            batch_size = 10
            for i in range(0, len(trade_dates), batch_size):
                batch_trade_dates = trade_dates[i:i + batch_size]
                batch_df = pd.DataFrame()

                # 遍历每个交易日
                for trade_date in batch_trade_dates:
                    if websocket:
                        await websocket.send_progress(
                            current=i + batch_trade_dates.index(trade_date) + 1,
                            total=len(trade_dates),
                            message=f"正在获取交易日 {trade_date} 的筹码分布数据..."
                        )

                    try:
                        # 调用Tushare接口获取该交易日所有个股的筹码分布数据
                        df = self.pro.cyq_perf(
                            trade_date=trade_date
                        )

                        if not df.empty:
                            # 将数据添加到批量DataFrame
                            batch_df = pd.concat([batch_df, df], ignore_index=True)
                    except Exception as e:
                        logger.error(f"Error fetching chip data for date {trade_date}: {str(e)}")
                        continue

                # 批量写入数据库
                if not batch_df.empty:
                    @sync_to_async
                    def bulk_create_chip_data(df):
                        with transaction.atomic():
                            # 获取现有的记录键值对
                            existing_records = {
                                (record.ts_code, record.trade_date): record 
                                for record in StockChipDistribution.objects.filter(
                                    trade_date__in=batch_df['trade_date'].unique().tolist()
                                )
                            }

                            # 准备批量创建和更新的数据
                            to_create = []
                            to_update = []

                            for _, row in df.iterrows():
                                key = (row['ts_code'], row['trade_date'])
                                if key in existing_records:
                                    record = existing_records[key]
                                    record.his_low = row['his_low']
                                    record.his_high = row['his_high']
                                    record.cost_5pct = row['cost_5pct']
                                    record.cost_15pct = row['cost_15pct']
                                    record.cost_50pct = row['cost_50pct']
                                    record.cost_85pct = row['cost_85pct']
                                    record.cost_95pct = row['cost_95pct']
                                    record.weight_avg = row['weight_avg']
                                    record.winner_rate = row['winner_rate']
                                    to_update.append(record)
                                else:
                                    to_create.append(StockChipDistribution(
                                        ts_code=row['ts_code'],
                                        trade_date=row['trade_date'],
                                        his_low=row['his_low'],
                                        his_high=row['his_high'],
                                        cost_5pct=row['cost_5pct'],
                                        cost_15pct=row['cost_15pct'],
                                        cost_50pct=row['cost_50pct'],
                                        cost_85pct=row['cost_85pct'],
                                        cost_95pct=row['cost_95pct'],
                                        weight_avg=row['weight_avg'],
                                        winner_rate=row['winner_rate']
                                    ))

                            # 批量创建新记录
                            created_count = 0
                            if to_create:
                                StockChipDistribution.objects.bulk_create(to_create, batch_size=1000)
                                created_count = len(to_create)

                            # 批量更新现有记录
                            updated_count = 0
                            if to_update:
                                StockChipDistribution.objects.bulk_update(
                                    to_update,
                                    ['his_low', 'his_high', 'cost_5pct', 'cost_15pct', 'cost_50pct',
                                     'cost_85pct', 'cost_95pct', 'weight_avg', 'winner_rate'],
                                    batch_size=1000
                                )
                                updated_count = len(to_update)

                            return created_count + updated_count

                    count = await bulk_create_chip_data(batch_df)
                    count_all += count
                    logger.info(f"Batch updated {count} chip distribution records for dates {batch_trade_dates}")
            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['STOCK_CHIP_DIST'],
                dt.datetime.strptime(trade_dates[-1], '%Y%m%d').date()
            )

            logger.info(f"Successfully updated {count_all} chip distribution records")
            return count_all
        except Exception as e:
            logger.error(f"Error updating stock chip distribution data: {str(e)}")
            raise

    

    async def batch_update_stock_margin(self, start_date=None, end_date=None, websocket=None):
        """
        批量更新市场和个股融资融券数据
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :param websocket: WebSocket对象，可选
        :return: 更新的记录数
        """
        logger.info(f'Starting update of stock margin data from {start_date} to {end_date}')
        try:
            # 解析日期
            if start_date and end_date:
                if '-' in start_date:  # 格式为YYYY-MM-DD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                # 如果没有指定日期范围，默认更新最近10个交易日
                end_date_obj = dt.date.today()
                start_date_obj = end_date_obj - dt.timedelta(days=20)  # 多取几天以确保包含10个交易日

            # 获取交易日列表
            trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
                cal_date__gte=start_date_obj.strftime('%Y%m%d'),
                cal_date__lte=end_date_obj.strftime('%Y%m%d'),
                is_open=True
            ).order_by('cal_date').values_list('cal_date', flat=True))

            if not trade_dates:
                logger.warning("No trading days found in the specified date range")
                return 0

            # 初始化计数器
            count_all = 0

            ######################################## 1.更新市场融资融券数据 ##############################################
            if websocket:
                await websocket.send_progress(
                    current=0,
                    total=len(trade_dates),
                    message="正在获取市场融资融券数据..."
                )

            # 调用Tushare接口获取市场融资融券数据
            try:
                market_df = self.pro.margin(
                    start_date=start_date,
                    end_date=end_date
                )

                if not market_df.empty:
                    @sync_to_async
                    def bulk_create_market_data(df):
                        with transaction.atomic():
                            # 获取现有的记录键值对
                            existing_records = {
                                (record.trade_date, record.exchange_id): record 
                                for record in MarketRzrq.objects.filter(
                                    trade_date__in=df['trade_date'].unique().tolist()
                                )
                            }

                            # 准备批量创建和更新的数据
                            to_create = []
                            to_update = []

                            for _, row in df.iterrows():
                                key = (row['trade_date'], row['exchange_id'])
                                if key in existing_records:
                                    record = existing_records[key]
                                    record.rzye = row['rzye']
                                    record.rzmre = row['rzmre']
                                    record.rzche = row['rzche']
                                    record.rqye = row['rqye']
                                    record.rqmcl = row['rqmcl']
                                    record.rzrqye = row['rzrqye']
                                    record.rqyl = row['rqyl']
                                    to_update.append(record)
                                else:
                                    to_create.append(MarketRzrq(
                                        trade_date=row['trade_date'],
                                        exchange_id=row['exchange_id'],
                                        rzye=row['rzye'],
                                        rzmre=row['rzmre'],
                                        rzche=row['rzche'],
                                        rqye=row['rqye'],
                                        rqmcl=row['rqmcl'],
                                        rzrqye=row['rzrqye'],
                                        rqyl=row['rqyl']
                                    ))

                            # 批量创建新记录
                            created_count = 0
                            if to_create:
                                MarketRzrq.objects.bulk_create(to_create, batch_size=1000)
                                created_count = len(to_create)

                            # 批量更新现有记录
                            updated_count = 0
                            if to_update:
                                MarketRzrq.objects.bulk_update(
                                    to_update,
                                    ['rzye', 'rzmre', 'rzche', 'rqye', 'rqmcl', 'rzrqye', 'rqyl'],
                                    batch_size=1000
                                )
                                updated_count = len(to_update)

                            return created_count + updated_count

                    count = await bulk_create_market_data(market_df)
                    count_all += count
                    logger.info(f"Updated {count} market margin records")
            except Exception as e:
                logger.error(f"Error fetching market margin data: {str(e)}")

            ######################################## 2.更新个股融资融券数据 ##############################################
            # 按每100个交易日分批处理
            batch_size = 100
            for i in range(0, len(trade_dates), batch_size):
                batch_trade_dates = trade_dates[i:i + batch_size]
                batch_df = pd.DataFrame()

                # 遍历每个交易日
                for trade_date in batch_trade_dates:
                    if websocket:
                        await websocket.send_progress(
                            current=i + batch_trade_dates.index(trade_date) + 1,
                            total=len(trade_dates),
                            message=f"正在获取交易日 {trade_date} 的个股融资融券数据..."
                        )

                    try:
                        # 调用Tushare接口获取该交易日的个股融资融券数据
                        df = self.pro.margin_detail(
                            trade_date=trade_date
                        )

                        if not df.empty:
                            # 将数据添加到批量DataFrame
                            batch_df = pd.concat([batch_df, df], ignore_index=True)
                    except Exception as e:
                        logger.error(f"Error fetching stock margin data for date {trade_date}: {str(e)}")
                        continue

                # 批量写入数据库
                if not batch_df.empty:
                    @sync_to_async
                    def bulk_create_stock_data(df):
                        with transaction.atomic():
                            # 获取现有的记录键值对
                            existing_records = {
                                (record.trade_date, record.ts_code): record 
                                for record in Stock_Rzrq.objects.filter(
                                    trade_date__in=df['trade_date'].unique().tolist()
                                )
                            }

                            # 准备批量创建和更新的数据
                            to_create = []
                            to_update = []

                            for _, row in df.iterrows():
                                key = (row['trade_date'], row['ts_code'])
                                if key in existing_records:
                                    record = existing_records[key]
                                    record.rzye = row['rzye']
                                    record.rqye = row['rqye']
                                    record.rzmre = row['rzmre']
                                    record.rqyl = row['rqyl']
                                    record.rzche = row['rzche']
                                    record.rqchl = row['rqchl']
                                    record.rqmcl = row['rqmcl']
                                    record.rzrqye = row['rzrqye']
                                    to_update.append(record)
                                else:
                                    to_create.append(Stock_Rzrq(
                                        trade_date=row['trade_date'],
                                        ts_code=row['ts_code'],
                                        rzye=row['rzye'],
                                        rqye=row['rqye'],
                                        rzmre=row['rzmre'],
                                        rqyl=row['rqyl'],
                                        rzche=row['rzche'],
                                        rqchl= row['rqchl'],
                                        rqmcl=row['rqmcl'],
                                        rzrqye=row['rzrqye']
                                    ))

                            # 批量创建新记录
                            created_count = 0
                            if to_create:
                                Stock_Rzrq.objects.bulk_create(to_create, batch_size=1000)
                                created_count = len(to_create)

                            # 批量更新现有记录
                            updated_count = 0
                            if to_update:
                                Stock_Rzrq.objects.bulk_update(
                                    to_update,
                                    ['rzye', 'rqye', 'rzmre', 'rqyl', 'rzche', 'rqchl', 'rqmcl', 'rzrqye'],
                                    batch_size=1000
                                )
                                updated_count = len(to_update)

                            return created_count + updated_count

                    count = await bulk_create_stock_data(batch_df)
                    count_all += count
                    logger.info(f"Batch updated {count} stock margin records for dates {batch_trade_dates}")

            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['MARKET_STOCK_RZRQ'],
                dt.datetime.strptime(trade_dates[-1], '%Y%m%d').date()
            )

            logger.info(f"Successfully updated {count_all} margin records")
            return count_all
        except Exception as e:
            logger.error(f"Error updating margin data: {str(e)}")
            raise

    async def batch_update_market_money_flow(self, start_date=None, end_date=None, websocket=None):
        """
        批量更新市场资金流向数据
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :param websocket: WebSocket对象，可选
        :return: 更新的记录数
        """
        logger.info(f'Starting update of market money flow data from {start_date} to {end_date}')
        try:
            # 解析日期
            if start_date and end_date:
                if '-' in start_date:  # 格式为YYYY-MM-DD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                # 如果没有指定日期范围，默认更新最近10个交易日
                end_date_obj = dt.date.today()
                start_date_obj = end_date_obj - dt.timedelta(days=20)  # 多取几天以确保包含10个交易日

            # 获取交易日列表
            trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
                cal_date__gte=start_date_obj.strftime('%Y%m%d'),
                cal_date__lte=end_date_obj.strftime('%Y%m%d'),
                is_open=True
            ).order_by('cal_date').values_list('cal_date', flat=True))

            if not trade_dates:
                logger.warning("No trading days found in the specified date range")
                return 0

            # 初始化计数器
            count_all = 0

            # 调用Tushare接口获取市场资金流向数据
            try:
                df = self.pro.moneyflow_mkt_dc(
                    start_date=start_date,
                    end_date=end_date
                )

                if not df.empty:
                    @sync_to_async
                    def bulk_create_market_money_flow(df):
                        with transaction.atomic():
                            # 获取现有的记录键值对
                            existing_records = {
                                record.trade_date: record 
                                for record in MarketMoneyFlowDC.objects.filter(
                                    trade_date__in=df['trade_date'].unique().tolist()
                                )
                            }

                            # 准备批量创建和更新的数据
                            to_create = []
                            to_update = []

                            for _, row in df.iterrows():
                                trade_date = row['trade_date']
                                if trade_date in existing_records:
                                    record = existing_records[trade_date]
                                    record.close_sh = row['close_sh']
                                    record.pct_change_sh = row['pct_change_sh']
                                    record.close_sz = row['close_sz']
                                    record.pct_change_sz = row['pct_change_sz']
                                    record.net_amount = row['net_amount']
                                    record.net_amount_rate = row['net_amount_rate']
                                    record.buy_elg_amount = row['buy_elg_amount']
                                    record.buy_elg_amount_rate = row['buy_elg_amount_rate']
                                    record.buy_lg_amount = row['buy_lg_amount']
                                    record.buy_lg_amount_rate = row['buy_lg_amount_rate']
                                    record.buy_md_amount = row['buy_md_amount']
                                    record.buy_md_amount_rate = row['buy_md_amount_rate']
                                    record.buy_sm_amount = row['buy_sm_amount']
                                    record.buy_sm_amount_rate = row['buy_sm_amount_rate']
                                   
                                    to_update.append(record)
                                else:
                                    to_create.append(MarketMoneyFlowDC(
                                        trade_date=trade_date,
                                        close_sh=row['close_sh'],
                                        pct_change_sh=row['pct_change_sh'],
                                        close_sz=row['close_sz'],
                                        pct_change_sz=row['pct_change_sz'],
                                        net_amount=row['net_amount'],
                                        net_amount_rate=row['net_amount_rate'],
                                        buy_elg_amount=row['buy_elg_amount'],
                                        buy_elg_amount_rate=row['buy_elg_amount_rate'],
                                        buy_lg_amount=row['buy_lg_amount'],
                                        buy_lg_amount_rate=row['buy_lg_amount_rate'],
                                        buy_md_amount=row['buy_md_amount'],
                                        buy_md_amount_rate=row['buy_md_amount_rate'],
                                        buy_sm_amount=row['buy_sm_amount'],
                                        buy_sm_amount_rate=row['buy_sm_amount_rate']
                                    ))

                            # 批量创建新记录
                            created_count = 0
                            if to_create:
                                MarketMoneyFlowDC.objects.bulk_create(to_create, batch_size=1000)
                                created_count = len(to_create)

                            # 批量更新现有记录
                            updated_count = 0
                            if to_update:
                                MarketMoneyFlowDC.objects.bulk_update(
                                    to_update,
                                    ['close_sh', 'pct_change_sh', 'close_sz', 'pct_change_sz', 'net_amount', 'net_amount_rate', 
                                     'buy_elg_amount', 'buy_elg_amount_rate', 'buy_lg_amount', 'buy_lg_amount_rate', 'buy_md_amount', 
                                     'buy_md_amount_rate', 'buy_sm_amount', 'buy_sm_amount_rate'],
                                    batch_size=1000
                                )
                                updated_count = len(to_update)

                            return created_count + updated_count

                    count = await bulk_create_market_money_flow(df)
                    count_all += count
                    logger.info(f"Updated {count} market money flow records")
            except Exception as e:
                logger.error(f"Error fetching market money flow data: {str(e)}")
                raise

            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['MARKET_MONEYFLOW_DC'],
                end_date_obj
            )

            logger.info(f"Successfully updated {count_all} market money flow records")
            return count_all
        except Exception as e:
            logger.error(f"Error updating market money flow data: {str(e)}")
            raise
    #实际上，ths个股资金流向数据从2024年12月24日开始才有数据。
    async def batch_update_stock_money_flow_ths(self, start_date=None, end_date=None, websocket=None):
        """
        批量更新个股资金流向数据
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :param websocket: WebSocket对象，可选
        :return: 更新的记录数
        """
        logger.info(f'Starting update of stock money flow data from {start_date} to {end_date}')
        try:
            # 解析日期
            if start_date and end_date:
                if '-' in start_date:  # 格式为YYYY-MM-DD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                # 如果没有指定日期范围，默认更新最近10个交易日
                end_date_obj = dt.date.today()
                start_date_obj = end_date_obj - dt.timedelta(days=20)  # 多取几天以确保包含10个交易日

            # 获取交易日列表
            trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
                cal_date__gte=start_date_obj.strftime('%Y%m%d'),
                cal_date__lte=end_date_obj.strftime('%Y%m%d'),
                is_open=True
            ).order_by('cal_date').values_list('cal_date', flat=True))

            if not trade_dates:
                logger.warning("No trading days found in the specified date range")
                return 0

            # 初始化计数器
            count_all = 0
            call_count = 0 # 调用接口的次数
            # 按每10个交易日分批处理
            batch_size = 10
            for i in range(0, len(trade_dates), batch_size):
                batch_trade_dates = trade_dates[i:i + batch_size]
                batch_df = pd.DataFrame()

                # 遍历每个交易日
                for trade_date in batch_trade_dates:
                    if call_count == 0:#表示重新一轮计时
                        start_time = time.time()#一次下载数据如果天数很多的话，可能需要等待多次，就需要在每次等待完成后重新开始计时，通过call_count来控制

                    if websocket:
                        await websocket.send_progress(
                            current=i + batch_trade_dates.index(trade_date) + 1,
                            total=len(trade_dates),
                            message=f"正在获取交易日 {trade_date} 的个股资金流向数据..."
                        )

                    try:
                        # 调用Tushare接口获取该交易日所有个股的资金流向数据
                        df = self.pro.moneyflow_ths(
                            trade_date=trade_date
                        )
                        call_count += 1
                        if call_count == TUSHARE_API_LIMIT['stock_moneyflow_ths_max_in_minute']:
                            total_time = time.time() - start_time
                            print(f'运行500次访问总共花费时间{total_time}秒')
                            if  total_time < SECONDS_IN_MINUTE:
                                await asyncio.sleep(SECONDS_IN_MINUTE - total_time)  
                                call_count = 0#需要清零，这样第二次还需要等待
                        if not df.empty:
                            # 将数据添加到批量DataFrame
                            batch_df = pd.concat([batch_df, df], ignore_index=True)
                    except Exception as e:
                        logger.error(f"Error fetching money flow data for date {trade_date}: {str(e)}")
                        continue

                # 批量写入数据库
                if not batch_df.empty:
                    @sync_to_async
                    def bulk_create_money_flow_data(df):
                        with transaction.atomic():
                            # 获取现有的记录键值对
                            existing_records = {
                                (record.ts_code, record.trade_date): record 
                                for record in StockMoneyFlowTHS.objects.filter(
                                    trade_date__in=batch_df['trade_date'].unique().tolist()
                                )
                            }

                            # 准备批量创建和更新的数据
                            to_create = []
                            to_update = []

                            for _, row in df.iterrows():
                                key = (row['ts_code'], row['trade_date'])
                                if key in existing_records:
                                    record = existing_records[key]
                                    record.name = row['name']
                                    record.pct_change = row['pct_change']
                                    record.latest = row['latest']
                                    record.net_amount = row['net_amount']
                                    record.net_d5_amount = row['net_d5_amount']
                                    record.buy_lg_amount = row['buy_lg_amount']
                                    record.buy_lg_amount_rate = row['buy_lg_amount_rate']
                                    record.buy_md_amount = row['buy_md_amount']
                                    record.buy_md_amount_rate = row['buy_md_amount_rate']
                                    record.buy_sm_amount = row['buy_sm_amount']
                                    record.buy_sm_amount_rate = row['buy_sm_amount_rate']
                                    to_update.append(record)
                                else:
                                    to_create.append(StockMoneyFlowTHS(
                                        ts_code=row['ts_code'],
                                        trade_date=row['trade_date'],
                                        name = row['name'],
                                        pct_change = row['pct_change'],
                                        latest = row['latest'],
                                        net_amount = row['net_amount'],
                                        net_d5_amount = row['net_d5_amount'],
                                        buy_lg_amount = row['buy_lg_amount'],
                                        buy_lg_amount_rate = row['buy_lg_amount_rate'],
                                        buy_md_amount = row['buy_md_amount'],
                                        buy_md_amount_rate = row['buy_md_amount_rate'],
                                        buy_sm_amount = row['buy_sm_amount'],
                                        buy_sm_amount_rate = row['buy_sm_amount_rate']
                                    ))

                            # 批量创建新记录
                            created_count = 0
                            if to_create:
                                StockMoneyFlowTHS.objects.bulk_create(to_create, batch_size=1000)
                                created_count = len(to_create)

                            # 批量更新现有记录
                            updated_count = 0
                            if to_update:
                                StockMoneyFlowTHS.objects.bulk_update(
                                    to_update,
                                    ['name', 'pct_change', 'latest', 'net_amount', 'net_d5_amount', 'buy_lg_amount', 
                                     'buy_lg_amount_rate', 'buy_md_amount', 'buy_md_amount_rate', 'buy_sm_amount', 'buy_sm_amount_rate'],
                                    batch_size=1000
                                )
                                updated_count = len(to_update)

                            return created_count + updated_count

                    count = await bulk_create_money_flow_data(batch_df)
                    count_all += count
                    logger.info(f"Batch updated {count} money flow records for dates {batch_trade_dates}")
            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['STOCK_MONEYFLOW_THS'],
                dt.datetime.strptime(trade_dates[-1], '%Y%m%d').date()
            )

            logger.info(f"Successfully updated {count_all} money flow records")
            return count_all
        except Exception as e:
            logger.error(f"Error updating stock money flow data: {str(e)}")
            raise

    async def batch_update_block_money_flow(self, start_date=None, end_date=None, websocket=None):
        """
        批量更新板块资金流向数据（概念+行业）
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :param websocket: WebSocket对象，可选
        :return: 更新的记录数
        """
        logger.info(f'Starting update of block money flow data from {start_date} to {end_date}')
        try:
            # 解析日期
            if start_date and end_date:
                if '-' in start_date:  # 格式为YYYY-MM-DD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                # 如果没有指定日期范围，默认更新最近10个交易日
                end_date_obj = dt.date.today()
                start_date_obj = end_date_obj - dt.timedelta(days=20)  # 多取几天以确保包含10个交易日

            start_date_obj = max(start_date_obj, dt.date(2024, 9, 1))  # 确保开始日期不早于2024年9月1日,更早日期都是空的。
            # 获取交易日列表
            trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
                cal_date__gte=start_date_obj.strftime('%Y%m%d'),
                cal_date__lte=end_date_obj.strftime('%Y%m%d'),
                is_open=True
            ).order_by('cal_date').values_list('cal_date', flat=True))

            if not trade_dates:
                logger.warning("No trading days found in the specified date range")
                return 0

            # 初始化计数器
            count_all = 0

            # 按每10个交易日分批处理
            batch_size = 10
            for i in range(0, len(trade_dates), batch_size):
                batch_trade_dates = trade_dates[i:i + batch_size]
                batch_df_gn = pd.DataFrame()  # 概念板块数据
                batch_df_hy = pd.DataFrame()  # 行业板块数据
                logger.info(f"batch_trade_dates = {batch_trade_dates}")
                # 遍历每个交易日
                for trade_date in batch_trade_dates:
                    if websocket:
                        await websocket.send_progress(
                            current=i + batch_trade_dates.index(trade_date) + 1,
                            total=len(trade_dates),
                            message=f"正在获取交易日 {trade_date} 的板块资金流向数据..."
                        )

                    try:
                        # 获取概念板块资金流向数据
                        df_gn = self.pro.moneyflow_cnt_ths(
                            trade_date=trade_date
                        )
                        if not df_gn.empty:
                            batch_df_gn = pd.concat([batch_df_gn, df_gn], ignore_index=True)
                        logger.info(f"Successfully fetched concept block money flow data for date {trade_date}")
                        # 获取行业板块资金流向数据
                        df_hy = self.pro.moneyflow_ind_ths(
                            trade_date=trade_date
                        )
                        if not df_hy.empty:
                            batch_df_hy = pd.concat([batch_df_hy, df_hy], ignore_index=True)
                    except Exception as e:
                        logger.error(f"Error fetching block money flow data for date {trade_date}: {str(e)}")
                        continue

                # 批量写入概念板块数据
                if not batch_df_gn.empty:
                    @sync_to_async
                    def bulk_create_gn_data(df):
                        with transaction.atomic():
                            # 获取现有的记录键值对
                            existing_records = {
                                (record.ts_code, record.trade_date): record 
                                for record in BlockMoneyFlowTHS_GN.objects.filter(
                                    trade_date__in=df['trade_date'].unique().tolist()
                                )
                            }

                            # 准备批量创建和更新的数据
                            to_create = []
                            to_update = []

                            for _, row in df.iterrows():
                                key = (row['ts_code'], row['trade_date'])
                                if row['ts_code'] is None:
                                    continue
                                if key in existing_records:
                                    record = existing_records[key]
                                    record.name = row['name']
                                    record.lead_stock = row['lead_stock']
                                    record.close_price = row['close_price']
                                    record.pct_change = row['pct_change']
                                    record.industry_index = row['industry_index']
                                    record.company_num = row['company_num']
                                    record.pct_change_stock = row['pct_change_stock']
                                    record.net_buy_amount = row['net_buy_amount']
                                    record.net_sell_amount = row['net_sell_amount']
                                    record.net_amount = row['net_amount']
                                    to_update.append(record)
                                else:
                                    to_create.append(BlockMoneyFlowTHS_GN(
                                        ts_code=row['ts_code'],
                                        trade_date=row['trade_date'],
                                        name=row['name'],
                                        lead_stock=row['lead_stock'],
                                        close_price=row['close_price'],
                                        pct_change=row['pct_change'],
                                        industry_index=row['industry_index'],
                                        company_num=row['company_num'],
                                        pct_change_stock=row['pct_change_stock'],
                                        net_buy_amount=row['net_buy_amount'],
                                        net_sell_amount=row['net_sell_amount'],
                                        net_amount=row['net_amount']
                                    ))

                            # 批量创建新记录
                            created_count = 0
                            if to_create:
                                BlockMoneyFlowTHS_GN.objects.bulk_create(to_create, batch_size=1000)
                                created_count = len(to_create)

                            # 批量更新现有记录
                            updated_count = 0
                            if to_update:
                                BlockMoneyFlowTHS_GN.objects.bulk_update(
                                    to_update,
                                    ['name', 'lead_stock', 'close_price', 'pct_change',
                                     'industry_index', 'company_num', 'pct_change_stock', 'net_buy_amount',
                                     'net_sell_amount', 'net_amount'],
                                    batch_size=1000
                                )
                                updated_count = len(to_update)

                            return created_count + updated_count

                    count_gn = await bulk_create_gn_data(batch_df_gn)
                    count_all += count_gn
                    logger.info(f"Batch updated {count_gn} concept block money flow records for dates {batch_trade_dates}")

                # 批量写入行业板块数据
                if not batch_df_hy.empty:
                    @sync_to_async
                    def bulk_create_hy_data(df):
                        with transaction.atomic():
                            # 获取现有的记录键值对
                            existing_records = {
                                (record.ts_code, record.trade_date): record 
                                for record in BlockMoneyFlowTHS_HY.objects.filter(
                                    trade_date__in=df['trade_date'].unique().tolist()
                                )
                            }

                            # 准备批量创建和更新的数据
                            to_create = []
                            to_update = []

                            for _, row in df.iterrows():
                                key = (row['ts_code'], row['trade_date'])
                                if key in existing_records:
                                    record = existing_records[key]
                                    record.industry = row['industry']
                                    record.lead_stock = row['lead_stock']
                                    record.close = row['close']
                                    record.pct_change = row['pct_change']
                                    record.company_num = row['company_num']
                                    record.pct_change_stock = row['pct_change_stock']
                                    record.close_price = row['close_price']
                                    record.net_buy_amount = row['net_buy_amount']
                                    record.net_sell_amount = row['net_sell_amount']
                                    record.net_amount = row['net_amount']
                                    to_update.append(record)
                                else:
                                    to_create.append(BlockMoneyFlowTHS_HY(
                                        ts_code=row['ts_code'],
                                        trade_date=row['trade_date'],
                                        industry=row['industry'],
                                        lead_stock=row['lead_stock'],
                                        close=row['close'],
                                        pct_change=row['pct_change'],
                                        company_num=row['company_num'],
                                        pct_change_stock=row['pct_change_stock'],
                                        close_price=row['close_price'],
                                        net_buy_amount=row['net_buy_amount'],
                                        net_sell_amount=row['net_sell_amount'],
                                        net_amount=row['net_amount']
                                    ))

                            # 批量创建新记录
                            created_count = 0
                            if to_create:
                                BlockMoneyFlowTHS_HY.objects.bulk_create(to_create, batch_size=1000)
                                created_count = len(to_create)

                            # 批量更新现有记录
                            updated_count = 0
                            if to_update:
                                BlockMoneyFlowTHS_HY.objects.bulk_update(
                                    to_update,
                                    ['industry', 'lead_stock', 'close', 'pct_change', 'company_num',
                                     'pct_change_stock', 'close_price', 'net_buy_amount', 'net_sell_amount',
                                     'net_amount'],
                                    batch_size=1000
                                )
                                updated_count = len(to_update)

                            return created_count + updated_count

                    count_hy = await bulk_create_hy_data(batch_df_hy)
                    count_all += count_hy
                    logger.info(f"Batch updated {count_hy} industry block money flow records for dates {batch_trade_dates}")

            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['BLOCK_MONEYFLOW_THS'],
                dt.datetime.strptime(trade_dates[-1], '%Y%m%d').date()
            )

            logger.info(f"Successfully updated {count_all} block money flow records (concept + industry)")
            return count_all
        except Exception as e:
            logger.error(f"Error updating block money flow data: {str(e)}")
            raise


    async def batch_update_stock_money_flow(self, start_date=None, end_date=None, websocket=None):
        """
        批量更新个股资金流向数据
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :param websocket: WebSocket对象，可选
        :return: 更新的记录数
        """
        logger.info(f'Starting update of stock money flow data from {start_date} to {end_date}')
        try:
            # 解析日期
            if start_date and end_date:
                if '-' in start_date:  # 格式为YYYY-MM-DD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                # 如果没有指定日期范围，默认更新最近10个交易日
                end_date_obj = dt.date.today()
                start_date_obj = end_date_obj - dt.timedelta(days=20)  # 多取几天以确保包含10个交易日

            # 获取交易日列表
            trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
                cal_date__gte=start_date_obj.strftime('%Y%m%d'),
                cal_date__lte=end_date_obj.strftime('%Y%m%d'),
                is_open=True
            ).order_by('cal_date').values_list('cal_date', flat=True))

            if not trade_dates:
                logger.warning("No trading days found in the specified date range")
                return 0

            # 初始化计数器
            count_all = 0

            # 按每10个交易日分批处理
            batch_size = 10
            for i in range(0, len(trade_dates), batch_size):
                batch_trade_dates = trade_dates[i:i + batch_size]
                batch_df = pd.DataFrame()

                # 遍历每个交易日
                for trade_date in batch_trade_dates:
                    if websocket:
                        await websocket.send_progress(
                            current=i + batch_trade_dates.index(trade_date) + 1,
                            total=len(trade_dates),
                            message=f"正在获取交易日 {trade_date} 的个股资金流向数据..."
                        )

                    try:
                        # 调用Tushare接口获取该交易日所有个股的资金流向数据
                        df = self.pro.moneyflow(
                            trade_date=trade_date
                        )

                        if not df.empty:
                            # 将数据添加到批量DataFrame
                            batch_df = pd.concat([batch_df, df], ignore_index=True)
                    except Exception as e:
                        logger.error(f"Error fetching money flow data for date {trade_date}: {str(e)}")
                        continue

                # 批量写入数据库
                if not batch_df.empty:
                    @sync_to_async
                    def bulk_create_money_flow_data(df):
                        with transaction.atomic():
                            # 获取现有的记录键值对
                            existing_records = {
                                (record.ts_code, record.trade_date): record 
                                for record in StockMoneyFlow.objects.filter(
                                    trade_date__in=batch_df['trade_date'].unique().tolist()
                                )
                            }

                            # 准备批量创建和更新的数据
                            to_create = []
                            to_update = []

                            for _, row in df.iterrows():
                                key = (row['ts_code'], row['trade_date'])
                                if key in existing_records:
                                    record = existing_records[key]
                                    record.buy_sm_vol = row['buy_sm_vol']
                                    record.buy_sm_amount = row['buy_sm_amount']
                                    record.sell_sm_vol = row['sell_sm_vol']
                                    record.sell_sm_amount = row['sell_sm_amount']
                                    record.buy_md_vol = row['buy_md_vol']
                                    record.buy_md_amount = row['buy_md_amount']
                                    record.sell_md_vol = row['sell_md_vol']
                                    record.sell_md_amount = row['sell_md_amount']
                                    record.buy_lg_vol = row['buy_lg_vol']
                                    record.buy_lg_amount = row['buy_lg_amount']
                                    record.sell_lg_vol = row['sell_lg_vol']
                                    record.sell_lg_amount = row['sell_lg_amount']
                                    record.buy_elg_vol = row['buy_elg_vol']
                                    record.buy_elg_amount = row['buy_elg_amount']
                                    record.sell_elg_vol = row['sell_elg_vol']
                                    record.sell_elg_amount = row['sell_elg_amount']
                                    record.net_mf_vol = row['net_mf_vol']
                                    record.net_mf_amount = row['net_mf_amount']
                                    to_update.append(record)
                                else:
                                    to_create.append(StockMoneyFlow(
                                        ts_code=row['ts_code'],
                                        trade_date=row['trade_date'],
                                        buy_sm_vol=row['buy_sm_vol'],
                                        buy_sm_amount=row['buy_sm_amount'],
                                        sell_sm_vol=row['sell_sm_vol'],
                                        sell_sm_amount=row['sell_sm_amount'],
                                        buy_md_vol=row['buy_md_vol'],
                                        buy_md_amount=row['buy_md_amount'],
                                        sell_md_vol=row['sell_md_vol'],
                                        sell_md_amount=row['sell_md_amount'],
                                        buy_lg_vol=row['buy_lg_vol'],
                                        buy_lg_amount=row['buy_lg_amount'],
                                        sell_lg_vol=row['sell_lg_vol'],
                                        sell_lg_amount=row['sell_lg_amount'],
                                        buy_elg_vol=row['buy_elg_vol'],
                                        buy_elg_amount=row['buy_elg_amount'],
                                        sell_elg_vol=row['sell_elg_vol'],
                                        sell_elg_amount=row['sell_elg_amount'],
                                        net_mf_vol=row['net_mf_vol'],
                                        net_mf_amount=row['net_mf_amount']
                                    ))

                            # 批量创建新记录
                            created_count = 0
                            if to_create:
                                StockMoneyFlow.objects.bulk_create(to_create, batch_size=1000)
                                created_count = len(to_create)

                            # 批量更新现有记录
                            updated_count = 0
                            if to_update:
                                StockMoneyFlow.objects.bulk_update(
                                    to_update,
                                    ['buy_sm_vol', 'buy_sm_amount', 'sell_sm_vol', 'sell_sm_amount',
                                        'buy_md_vol', 'buy_md_amount', 'sell_md_vol', 'sell_md_amount',
                                        'buy_lg_vol', 'buy_lg_amount', 'sell_lg_vol', 'sell_lg_amount',
                                        'buy_elg_vol', 'buy_elg_amount', 'sell_elg_vol', 'sell_elg_amount',
                                        'net_mf_vol', 'net_mf_amount'],
                                    batch_size=1000
                                )
                                updated_count = len(to_update)

                            return created_count + updated_count

                    count = await bulk_create_money_flow_data(batch_df)
                    count_all += count
                    logger.info(f"Batch updated {count} money flow records for dates {batch_trade_dates}")
            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['STOCK_MONEYFLOW'],
                dt.datetime.strptime(trade_dates[-1], '%Y%m%d').date()
            )

            logger.info(f"Successfully updated {count_all} money flow records")
            return count_all
        except Exception as e:
            logger.error(f"Error updating stock money flow data: {str(e)}")
            raise